package com.yifeng.repo.flink.data.transport.streaming.connectors.mysql.antlr.listener;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.yifeng.repo.flink.data.transport.dto.DdlTypeEnums;

import io.debezium.connector.mysql.antlr.MySqlAntlrDdlParser;
import io.debezium.ddl.parser.mysql.generated.MySqlParser;
import io.debezium.ddl.parser.mysql.generated.MySqlParser.UidContext;
import io.debezium.ddl.parser.mysql.generated.MySqlParserBaseListener;
import io.debezium.relational.TableId;

/**
 * mysql的ddl解析监听类
 * @author lijing
 * @since 2023-04-14
 *
 */
public class MysqlDdlParserListener extends MySqlParserBaseListener {

    private final static Logger LOG = LoggerFactory.getLogger(MysqlDdlParserListener.class);

    private final MySqlAntlrDdlParser parser;
    
    /**
     * 表名
     */
    private String tableName;
    
	/**
	 * key: ddl变更的类型
	 * Map : rename的前后字段名称，key为老名称, value为新名称
	 */
	private Map<String, Map<String,String>> renameMap;
	
	/**
	 * key: ddl变更的类型
	 * value: alter table 新增、修改、删除的字段
	 */
	private Map<String, List<String>> columnMap;
	
	public MysqlDdlParserListener(String tableName) {
		super();
		this.parser = new MySqlAntlrDdlParser();
		this.tableName = tableName;
		this.columnMap = new HashMap<String, List<String>>();
		this.renameMap = new HashMap<String, Map<String,String>>();
	}


    @Override
    public void enterAlterByAddColumn(MySqlParser.AlterByAddColumnContext ctx) {
    	doEnterAlterByAddColumns(ctx.uid());
        super.exitAlterByAddColumn(ctx);
    }

    @Override
    public void enterAlterByAddColumns(MySqlParser.AlterByAddColumnsContext ctx) {
    	doEnterAlterByAddColumns(ctx.uid());
        super.enterAlterByAddColumns(ctx);
    }
    
    private void doEnterAlterByAddColumns(List<UidContext> uidList) {
    	List<String> columnList = new ArrayList<String>();
		for (MySqlParser.UidContext uidContext : uidList) {
    		columnList.add(parser.parseName(uidContext));
        }
		if(columnMap.get(DdlTypeEnums.ADD.getType()) != null) {
			columnList.addAll(columnMap.get(DdlTypeEnums.ADD.getType()));
		}
		columnMap.put(DdlTypeEnums.ADD.getType(), columnList);
    }

    @Override
    public void enterAlterByChangeColumn(MySqlParser.AlterByChangeColumnContext ctx) {
		Map<String,String> map = new HashMap<String,String>();
		map.put(parser.parseName(ctx.oldColumn), parser.parseName(ctx.newColumn));
		if(renameMap.get(DdlTypeEnums.RENAME_COLUMN.getType()) != null) {
			map.putAll(renameMap.get(DdlTypeEnums.RENAME_COLUMN.getType()));
		}
		renameMap.put(DdlTypeEnums.RENAME_COLUMN.getType(), map);
        super.enterAlterByChangeColumn(ctx);
    }

    @Override
    public void enterAlterByModifyColumn(MySqlParser.AlterByModifyColumnContext ctx) {
		List<String> columnList = new ArrayList<String>();
		for (MySqlParser.UidContext uidContext : ctx.uid()) {
    		columnList.add(parser.parseName(uidContext));
        }
    	if(columnMap.get(DdlTypeEnums.MODIFY.getType()) != null) {
			columnList.addAll(columnMap.get(DdlTypeEnums.MODIFY.getType()));
		}
		columnMap.put(DdlTypeEnums.MODIFY.getType(), columnList);
        super.enterAlterByModifyColumn(ctx);
    }

    @Override
    public void enterAlterByDropColumn(MySqlParser.AlterByDropColumnContext ctx) {
		List<String> columnList = new ArrayList<String>();
    	columnList.add(parser.parseName(ctx.uid()));
    	if(columnMap.get(DdlTypeEnums.DROP.getType()) != null) {
			columnList.addAll(columnMap.get(DdlTypeEnums.DROP.getType()));
		}
		columnMap.put(DdlTypeEnums.DROP.getType(), columnList);
        super.enterAlterByDropColumn(ctx);
    }

    @Override
    public void enterAlterByRename(MySqlParser.AlterByRenameContext ctx) {
		final TableId newTableId = ctx.uid() != null
                ? parser.resolveTableId(parser.currentSchema(), parser.parseName(ctx.uid()))
                : parser.parseQualifiedTableId(ctx.fullId());
        Map<String,String> map = new HashMap<String,String>();
        map.put(tableName, newTableId.table());
		renameMap.put(DdlTypeEnums.RENAME_TABLE.getType(), map);
        super.enterAlterByRename(ctx);
    }

    @Override
    public void enterAlterByRenameColumn(MySqlParser.AlterByRenameColumnContext ctx) {
		Map<String,String> map = new HashMap<String,String>();
		map.put(parser.parseName(ctx.oldColumn), parser.parseName(ctx.newColumn));
		if(renameMap.get(DdlTypeEnums.RENAME_COLUMN.getType()) != null) {
			map.putAll(renameMap.get(DdlTypeEnums.RENAME_COLUMN.getType()));
		}
		renameMap.put(DdlTypeEnums.RENAME_COLUMN.getType(), map);
        super.enterAlterByRenameColumn(ctx);
    }


	public Map<String, Map<String, String>> getRenameMap() {
		return renameMap;
	}


	public void setRenameMap(Map<String, Map<String, String>> renameMap) {
		this.renameMap = renameMap;
	}


	public Map<String, List<String>> getColumnMap() {
		return columnMap;
	}


	public void setColumnMap(Map<String, List<String>> columnMap) {
		this.columnMap = columnMap;
	}

}
